/**
 * Copyright 2013 ABSir's Studio
 * 
 * All right reserved
 *
 * Create on 2013-10-22 下午2:02:53
 */
package com.absir.server.socket;

import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.absir.context.core.ContextFactory;
import com.absir.context.core.ContextUtils;
import com.absir.core.kernel.KernelReflect;
import com.absir.server.exception.ServerException;
import com.absir.server.exception.ServerStatus;
import com.absir.server.socket.resolver.SocketChannelResolver;

/**
 * @author absir
 * 
 */
@SuppressWarnings({ "rawtypes" })
public class SocketServer {

	/** LOGGER */
	protected static final Logger LOGGER = LoggerFactory.getLogger(SocketServer.class);

	/** globalSelector */
	private static ConcurrentHashMap<SocketChannel, SelSession> globalSelSessionMap;

	/**
	 * 开启全局selSessionMap
	 * 
	 * @throws IOException
	 * 
	 */
	protected void startSelSessionMap() throws IOException {
		synchronized (SocketServer.class) {
			if (globalSelSessionMap == null) {
				globalSelSessionMap = new ConcurrentHashMap<SocketChannel, SelSession>();
				Thread globalSelectorThread = new Thread() {

					/*
					 * (non-Javadoc)
					 * 
					 * @see java.lang.Thread#run()
					 */
					@Override
					public void run() {
						try {
							long sessionDelay = SocketServerContext.getSessionDelay();
							Iterator<Entry<SocketChannel, SelSession>> iterator;
							Entry<SocketChannel, SelSession> entry;
							SelSession selSession;
							SocketChannel socketChannel;
							long contextTime;
							while (true) {
								Thread.sleep(sessionDelay);
								contextTime = System.currentTimeMillis();
								iterator = globalSelSessionMap.entrySet().iterator();
								while (iterator.hasNext()) {
									entry = iterator.next();
									selSession = entry.getValue();
									socketChannel = entry.getKey();
									try {
										if (!ContextFactory.isWarnExecutor() && socketChannel.isConnected()) {
											if (selSession.getIdleMaxTime() > contextTime) {
												continue;

											} else {
												selSession.getSocketReceiver().idle(selSession, socketChannel, contextTime);
												if (selSession.getIdleMaxTime() > contextTime) {
													continue;
												}
											}
										}

										iterator.remove();
										sessionClose(selSession, socketChannel);

									} catch (Throwable e) {
										LOGGER.error("idle error " + socketChannel, e);
									}
								}

								iterator = null;
								entry = null;
								selSession = null;
								socketChannel = null;
							}

						} catch (InterruptedException e) {
						}
					}
				};

				globalSelectorThread.setName("SocketServer.globalSelectorThread");
				globalSelectorThread.setDaemon(true);
				globalSelectorThread.start();
			}
		}
	}

	/**
	 * @param selSession
	 * @param socketChannel
	 * @throws Throwable
	 */
	public static void sessionClose(final SelSession selSession, final SocketChannel socketChannel) throws Throwable {
		try {
			socketChannel.close();

		} catch (Exception e) {
		}

		if (selSession != null) {
			synchronized (selSession) {
				SocketBuffer socketBuffer = selSession.getSocketBuffer();
				if (socketBuffer != null) {
					final Serializable id = socketBuffer.getId();
					if (id != null && id != SocketReceiverContext.UN_REGISTER_ID) {
						socketBuffer.setId(SocketReceiverContext.UN_REGISTER_ID);
						ContextUtils.getThreadPoolExecutor().execute(new Runnable() {

							@Override
							public void run() {
								try {
									selSession.getSocketReceiver().unRegister(selSession, id, socketChannel);

								} catch (Throwable e) {
									LOGGER.error("session close " + socketChannel, e);
								}
							}
						});
					}
				}
			}
		}
	}

	/**
	 * @return
	 */
	public static int sessionCount() {
		return globalSelSessionMap.size();
	}

	/**
	 * @param socketChannel
	 * @return
	 */
	public static SelSession forSession(SocketChannel socketChannel) {
		return globalSelSessionMap == null ? null : globalSelSessionMap.get(socketChannel);
	}

	/**
	 * @param server
	 */
	public static void closeAllSocketChannelServer(SocketServer server) {
		Entry<SocketChannel, SelSession> entry;
		SelSession selSession;
		SocketChannel socketChannel;
		Iterator<Entry<SocketChannel, SelSession>> iterator = globalSelSessionMap.entrySet().iterator();
		while (iterator.hasNext()) {
			entry = iterator.next();
			selSession = entry.getValue();
			if (selSession.getSocketServer() == server) {
				socketChannel = entry.getKey();
				try {
					iterator.remove();
					sessionClose(selSession, socketChannel);
					// globalSelSessionMap.put(socketChannel, selSession);

				} catch (Throwable e) {
					LOGGER.error("idle error " + socketChannel, e);
				}
			}
		}
	}

	/** serverSocket */
	protected ServerSocketChannel serverSocketChannel;

	/** serverSelector */
	protected Selector serverSelector;

	/** port */
	protected int port;

	/** socketReceiver */
	protected SocketReceiver socketReceiver;

	/**
	 * @return the port
	 */
	public int getPort() {
		return port;
	}

	/**
	 * @return the socketReceiver
	 */
	public SocketReceiver getSocketReceiver() {
		return socketReceiver;
	}

	/**
	 * @return
	 */
	public boolean isClosed() {
		return serverSelector == null;
	}

	/**
	 * 开始服务
	 * 
	 * @param port
	 * @param backlog
	 * @param inetAddress
	 * @param bufferSize
	 * @param receiveBufferSize
	 * @param sendBufferSize
	 * @param channelResolver
	 * @param receiver
	 * @throws IOException
	 */
	public void start(int port, int backlog, InetAddress inetAddress, final int bufferSize, int receiveBufferSize,
			final int sendBufferSize, final SocketChannelResolver channelResolver, final SocketReceiver receiver)
			throws IOException {
		if (serverSocketChannel != null) {
			throw new ServerException(ServerStatus.ON_FAIL);
		}

		if (globalSelSessionMap == null) {
			startSelSessionMap();
		}

		this.port = port;
		LOGGER.info("start socket " + inetAddress + " port " + port);
		// 初始化监听服务
		serverSocketChannel = ServerSocketChannel.open();
		serverSocketChannel.configureBlocking(false);
		serverSocketChannel.socket().setReceiveBufferSize(receiveBufferSize);
		serverSocketChannel.socket().bind(new InetSocketAddress(inetAddress, port), backlog);

		// 接受请求
		serverSelector = Selector.open();
		serverSocketChannel.register(serverSelector, SelectionKey.OP_ACCEPT);
		socketReceiver = receiver;
		ContextUtils.getThreadPoolExecutor().execute(new Runnable() {

			@Override
			public void run() {
				Selector selector = serverSelector;
				ByteBuffer buffer = ByteBuffer.allocate(bufferSize);
				while (selector == serverSelector && !Thread.interrupted()) {
					try {
						selector.select();
						Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
						while (iterator.hasNext()) {
							SelectionKey key = iterator.next();
							iterator.remove();
							if (key.isAcceptable()) {
								// 接受请求
								SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
								socketChannel.configureBlocking(false);
								socketChannel.socket().setSendBufferSize(sendBufferSize);
								try {
									// 处理注册请求
									long acceptTimeout = receiver.acceptTimeout(socketChannel);
									if (acceptTimeout > 0) {
										SelSession selSession = new SelSession(SocketServer.this);
										selSession.retainIdleTimeout(acceptTimeout);
										globalSelSessionMap.put(socketChannel, selSession);
										socketChannel.register(selector, SelectionKey.OP_READ, selSession);
										continue;
									}

								} catch (Throwable e) {
									if (SocketServerContext.getCloseDebug()) {
										e.printStackTrace();
									}
								}

								close(socketChannel);

							} else {
								// 处理数据
								SelSession selSession = (SelSession) key.attachment();
								SocketBuffer socketBuffer = selSession.getSocketBuffer();
								SocketChannel socketChannel = (SocketChannel) key.channel();
								try {
									buffer.clear();
									int length = socketChannel.read(buffer);
									if (length > 0) {
										if (socketBuffer == null) {
											socketBuffer = channelResolver.createSocketBuff();
											selSession.setSocketBuffer(socketBuffer);
										}

										byte[] array = buffer.array();
										int position = 0;
										while (position < length) {
											position = channelResolver.readByteBuffer(socketBuffer, array, position, length);
											if (socketBuffer.getBuff() != null
													&& socketBuffer.getLength() <= socketBuffer.getBuffLengthIndex()) {
												if (socketBuffer.getId() == null) {
													receiver.register(socketChannel, selSession);
													Serializable id = socketBuffer.getId();
													if (id == null) {
														break;
													}

												} else {
													receiver.receiveByteBuffer(socketChannel, selSession);
												}

												channelResolver.readByteBufferDone(socketBuffer);
											}

											position++;
										}

										// 注销请求
										// key.cancel();
										continue;
									}

								} catch (Throwable e) {
									if (SocketServerContext.getCloseDebug()) {
										e.printStackTrace();
									}
								}

								// 注销请求
								key.cancel();
								try {
									if (socketBuffer != null) {
										globalSelSessionMap.remove(socketChannel);
										sessionClose(selSession, socketChannel);
									}

								} catch (Throwable e) {
									if (SocketServerContext.getCloseDebug()) {
										e.printStackTrace();
									}
								}

								close(socketChannel);
							}
						}

					} catch (Throwable e) {
						if (SocketServerContext.getCloseDebug()) {
							e.printStackTrace();
						}
					}
				}

				try {
					selector.close();

				} catch (Throwable e) {
				}

				close();
			}
		});
	}

	/**
	 * 关闭服务
	 */
	public void close() {
		LOGGER.info("stop socket port " + port);
		synchronized (this) {
			if (serverSocketChannel != null) {

				closeAllSocketChannelServer(SocketServer.this);
				try {
					serverSocketChannel.close();
					LOGGER.info("stop socket port " + port + " => " + serverSocketChannel.socket().getLocalPort());

				} catch (Throwable e) {
					e.printStackTrace();
				}

				serverSocketChannel = null;
				serverSelector = null;
			}
		}
	}

	/**
	 * @param socketChannel
	 */
	public static void close(SocketChannel socketChannel) {
		try {
			if (globalSelSessionMap != null && socketChannel.isRegistered()) {
				SelSession selSession = globalSelSessionMap.remove(socketChannel);
				if (selSession != null) {
					sessionClose(selSession, socketChannel);
				}
			}

			socketChannel.close();

		} catch (Throwable e) {
		}

		if (SocketServerContext.getCloseDebug()) {
			new Exception().printStackTrace();
		}
	}

	private static int selectionKeyOpt;

	private static Field keysField;

	/**
	 * @param socketChannel
	 * @return
	 */
	public static SelectionKey[] selectionKeys(SocketChannel socketChannel) {
		if (socketChannel.isRegistered()) {
			if (selectionKeyOpt < 0) {
				return null;
			}

			if (selectionKeyOpt == 0) {
				keysField = KernelReflect.declaredField(SocketChannel.class, "keys");
				if (keysField == null) {
					selectionKeyOpt = -1;
					return null;

				} else {
					selectionKeyOpt = 1;
				}
			}

			return (SelectionKey[]) KernelReflect.get(socketChannel, keysField);
		}

		return null;
	}

	/**
	 * @param socketChannel
	 * @return
	 */
	public static SelectionKey selectionKey(SocketChannel socketChannel) {
		SelectionKey[] keys = selectionKeys(socketChannel);
		return keys == null || keys.length == 0 ? null : keys[0];
	}
}
